package com.coolpad.MQ.kafka;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * 将消息发送到kafka队列中
 *
 * @author liming   16-4-27.
 */
public class ProducerHelpers {
    private static final Logger LOGGER = LoggerFactory.getLogger(com.coolpad.MQ.kafka.ProducerHelpers.class);
    private static com.coolpad.MQ.kafka.KafkaProducerWrapper<String,String> producer;

    static {
        producer = new com.coolpad.MQ.kafka.KafkaProducerWrapper<>();
    }

    public ProducerHelpers() {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.16.0.14:9092");
        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        LOGGER.info("initialize kafka producer. properties = %s", p);

        producer.init(p);

        producer.start();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            producer.close();
        }));
    }

    public boolean publish(String topic, String value, String key) {
        if (key == null) key = "";
        long start = System.currentTimeMillis();
        try {
            LOGGER.info("send {}, {}, {} to kafka", topic, value, key);
            boolean result = producer.send(topic, key, value);

            return result;
        } catch (Exception e) {
            LOGGER.info("publish message %s to kafka failed: ", value, e);
            return false;
        } finally {
        }
    }

}

